package com.atguigu.sparkstreaming.demos

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/7/15
 *
 *    编程入口(上下文Context)和编程模型:
 *
 *    编程入口:
 *        SparkCore :   SparkContext
 *        SparkSql  :   SparkSession(内置SparkContext)
 *        SparkStreaming :   StreamingContext(内置SparkContext)
 *
 *    编程模型:
 *        SparkCore :   RDD
 *         SparkSql  :   DataFrame DataSet
 *        SparkStreaming:  DStream
 *
 *
 *    StreamingContext: 编程的核心入口。
 *                    用来从多种数据源创建 DStream
 *
 *                怎么玩?
 *                      ①创建 StreamingContext
 *                      ②从StreamingContext 中获取 DStream
 *                      ③调用 DStream的算子(高度抽象原语)计算
 *                      ④以上3步都是懒加载的，什么时候真正开始运算
 *                          启动APP
 *                          StreamingContext.start()
 *                          停止:
 *                          StreamingContext.stop()
 *                          流式计算： 启动后，一定24h不停运算
 *                              StreamingContext.awaitTermination() 等待发停止信号 或 出现异常会停止
 *
 *     ---------------------------------------
 *      跟着官网做: https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html
 *
 *      如果消费的是Kafka主题中的数据，这个主题有几个分区，消费后的数据封装的RDD，就有几个partition
 *
 *          repartition: 重新分区
 */
object WordCountDemo {

  def main(args: Array[String]): Unit = {

    /*
        batchDuration: Duration:  一个批次的持续时间。
                                  采集多久的数据为1个批次

                         Milliseconds(毫秒数)
                         Seconds(秒)
                         Minutes(分钟)
     */



   /* val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCountDemo")

    val sparkContext = new SparkContext(sparkConf)

    val streamingContext1 = new StreamingContext(sparkContext, Seconds(10))

    val streamingContext = new StreamingContext(sparkConf, Seconds(10))*/

    //①创建 StreamingContext
    val streamingContext = new StreamingContext("local[*]", "WordCountDemo", Seconds(10))

    //如果想获取其中的SparkContext
    //val sparkContext: SparkContext = streamingContext.sparkContext
    // ②从StreamingContext 中获取 DStream,参考数据源(hdfs，kafka,TCP Port)
    /*
                fileStream:  流式读取HDFS目录中新增的文件
                socketStream:  流式读取 固定主机:port下发送的数据

     */

    /*
          App扮演的是消费者的角色。 只要是消费者都有消费者的参数设置
          必须有:
                bootstrap.servers
                key/value的deserializer
                group.id

                auto.offset.reset： 从主题的哪个位置消费
                    earliest:   如果组从来没有消费过主题，从主题的最早位置消费
                    latest:    如果组从没没有消费过主题，从主题的最后(当前，最新)位置消费
                    none:     如果组已经消费了，从earliest 转为none，从已经提交的offsets后继续消费。

                enable.auto.commit： 是否允许consumer自动提交offsets
                          如果使用的是kafka 0.10版本以上的版本，都是将offsets提交到 kafka内置的 _consumer_offsets中
     */
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "220309",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    /*
        要消费的主题。
          理论上允许一个流消费多种主题的数据。
          但是在操作时，只写一个主题！
              原因:  不同主题中保存的数据类型是不一样的。如果一个流消费了两种不同的数据，流中混杂了两种数据，
                      在进行处理时，都需要对数据类型判断，判断是否是要处理的某种类型，编程逻辑难于维护。杂乱。

          如果需要消费两个主题，应该每个主题一个流，获取两个流！ 每个流中只有一种数据！
     */
    val topics = Array("topicA")

    /*
        如何从kakfa数据源获取DStream。
            全是固定代码！

     def createDirectStream[K, V](
      ssc: StreamingContext,       程序入口
      locationStrategy: LocationStrategy, 位置策略:  kafka的broker和SparkApp的 Executor的位置关系(是不是同一个机器，同一个机架，同一个机房)
                                            调度Task到Executor时，有本地化(移动计算而不是移动数据)级别。

                                            如果当前要消费的TopicA的0号分区的Leader在Hadoop102机器，
                                            我这个App恰好在Hadoop102启动了一个Executor，那么这个Task就应该调度给102的Executor，不应该
                                            给其他的Executor

                                            99%都是   PreferConsistent

      consumerStrategy: ConsumerStrategy[K, V]： 消费策略。
                                      独立消费者:   明确告诉你要消费哪个主题的哪个分区，从哪个Offset开始消费。
                                                      Assign
                                      非独立消费者(99%):  告诉你要消费哪个主题。有Kafka集群自动给消费者组中的每个线程分配分区，
                                                      读取之前提交的offsets，从之前提交的位置取消费
                                                      Subscribe

             ConsumerRecord[K,V]: 从kafka消费到的一条数据。 只获取V

             ProducerRecord[K,V]:  V:封装data
                                   K： 封装meta data。用于分区等
                                          partiton=0

    )
     */
    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val ds1: DStream[String] = ds.map(record => record.value())

    // ds2中的 [String]是一个单词
    val ds2: DStream[String] = ds1.flatMap(line => line.split(" "))

    val ds3: DStream[(String, Int)] = ds2.map(word => (word, 1)).reduceByKey(_ + _)

    //输出 在屏幕打印  print():只打印前10行
    ds3.print(1000)

    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
